package com.group2.edu.realtime.dwd.log.split.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.constant.Constant;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import com.ibm.icu.util.Output;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Date;

/**
 * @author 高耀
 * @date 2024/12/14 15:15<p></p>
 */
public class DwdBaseLog extends BaseApp {
    public static void main(String[] args) {
        new DwdBaseLog().start(
                10011,
                4,
                "dwd_base_log",
                Constant.TOPIC_LOG
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> mainSource) {
        //TODO 1 将流中的JSONStr转成JSONObject，如果有脏数据，放在侧输出流中写入到kafka
        //1.1 定义脏数据侧输出流标签
        OutputTag<String> dirtyDataTag = new OutputTag<String>("dirty_data_tag") {
        };
        //1.2 转换处理
        SingleOutputStreamOperator<JSONObject> convertedDs = mainSource.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String line,
                                               ProcessFunction<String, JSONObject>.Context context,
                                               Collector<JSONObject> collector
                    ) throws Exception {
                        try {
                            //尝试转
                            JSONObject jsonObject = JSON.parseObject(line);
                            //转换成功说明不是脏数据
                            collector.collect(jsonObject);
                        } catch (Exception e) {
                            //产生异常说明是脏数据，放入侧输出流即可
                            context.output(dirtyDataTag, line);
                        }
                    }
                }
        );
        //1.3 取出侧输出流，写入kafka
        SideOutputDataStream<String> dirtyDataDs = convertedDs.getSideOutput(dirtyDataTag);
        //测试
//        dirtyDataDs.print("脏数据");
        convertedDs.print("正确数据");
        //写入脏数据所在的topic
        dirtyDataDs.sinkTo(FlinkSinkUtil.getKafkaSink("dirty_data"));


        //TODO 2 修正流中日志数据的is_new字段的值
        //2.1 先按用户分组,也就是设备id mid
        KeyedStream<JSONObject, String> keyedDs =
                convertedDs.keyBy(
                        jsonObject -> jsonObject.getJSONObject("common").getString("mid"));
        //2.2 使用process函数进行处理
        SingleOutputStreamOperator<JSONObject> fixedIsNewDs = keyedDs.process(
                /**
                 * 使用按键分区的process,
                 * 分组键类型是String，流中数据是JSONObject形式，写出的形式还是JSONObject
                 */
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    //定义一个按键分区的值状态，用来存每个用户上次登录的时间
                    private ValueState<String> lastViewDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //创建一个值状态描述器
                        ValueStateDescriptor<String> valueStateDesc =
                                new ValueStateDescriptor<>("valueStateDesc", String.class);
                        //对状态的 初始化 或 再恢复
                        lastViewDateState = getRuntimeContext().getState(valueStateDesc);
                    }

                    @Override
                    public void processElement(
                            JSONObject jsonObject,
                            KeyedProcessFunction<String, JSONObject, JSONObject>.Context context,
                            Collector<JSONObject> collector
                    ) throws Exception {
                        //取出数据中的is_new的值
                        String isNew = jsonObject.getJSONObject("common").getString("is_new");
                        //取出数据中的时间，转成日期的格式
                        String currentDate = DateFormatUtils.format(
                                jsonObject.getLong("ts"),
                                "yyyy-MM-dd"
                        );
                        //获取状态中日期
                        String lastViewDate = this.lastViewDateState.value();
                        //进行分类判断
                        if ("1".equals(isNew)) {
                            //本条数据显示为新用户
                            if (StringUtils.isNotEmpty(lastViewDate)) {
                                //状态中有数据，但可能是同一天的数据
                                if (!currentDate.equals(lastViewDate)) {
                                    //不是同一天的，说明确实不应该是新用户，修正数据为0
                                    jsonObject.getJSONObject("common").put("is_new", "0");
                                }
                            } else {
                                //状态中确实无记录，说明是新用户
                                //将最新日期添加到状态中
                                lastViewDateState.update(currentDate);
                            }
                        } else {
                            //说明不是新用户
                            if (StringUtils.isEmpty(lastViewDate)) {
                                //但是状态中无数据，
                                //可能是在数仓项目上线前就登录过，更新状态为前一天用来表示老用户
                                Date date = DateUtils.addDays(
                                        DateUtils.parseDate(currentDate, "yyyy-MM-dd"), -1);
                                lastViewDateState.update(DateFormatUtils.format(date, "yyyy-MM-dd"));
                            }
                        }
                        //写出本条数据
                        collector.collect(jsonObject);
                    }
                }
        );
        //测试
//        fixedIsNewDs.print();

        //TODO 3 分流操作：分出错误日志、启动日志、曝光日志、动作日志、播放日志，剩下的主流为页面日志，最后分别写入不同topic
        //3.1 定义四个侧输出流，直接定义为String类型方便直接写入Kafka
        //错误流
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        //启动流
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        //动作流
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        //曝光流
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};

        //播放流
        OutputTag<String> playTag = new OutputTag<String>("playTag") {};

        //3.2使用process方法进行分流操作
        SingleOutputStreamOperator<String> pageDs = fixedIsNewDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(
                            JSONObject jsonObject,
                            ProcessFunction<JSONObject, String>.Context context,
                            Collector<String> collector
                    ) throws Exception {
                        //所有类型日志都可能是错误日志，先选出错误日志
                        //尝试获取错误日志字段信息
                        JSONObject errObj = jsonObject.getJSONObject("err");
                        if (errObj != null) {
                            //~~~错误日志~~~
                            //是错误日志，将整条数据写入错误日志流
                            context.output(errTag, jsonObject.toJSONString());
                            //移除错误日志的相关字段
                            jsonObject.remove("err");
                        }

                        //尝试获取启动日志字段信息
                        JSONObject startObj = jsonObject.getJSONObject("start");
                        //尝试获取播放日志字段信息
                        JSONObject appVideoObj = jsonObject.getJSONObject("appVideo");

                        if (startObj != null) {
                            //~~~启动日志~~~
                            // 写入启动日志侧输出流即可
                            context.output(startTag, jsonObject.toJSONString());
                        } else if (appVideoObj != null) {
                            //~~~播放日志~~~
                            // 写入播放日志侧输出流即可
                            context.output(playTag, jsonObject.toJSONString());
                        } else {
                            //~~~页面日志~~~
                            //不是启动日志也不是播放日志就一定是页面日志，其下又分为动作日志和曝光日志，这两个字段会同时出现
                            //取出公共部分
                            JSONObject commonObj = jsonObject.getJSONObject("common");
                            JSONObject pageObj = jsonObject.getJSONObject("page");
                            Long ts = jsonObject.getLong("ts");

                            //~~~启动日志~~~
                            //取出启动日志字段集合
                            JSONArray actionArray = jsonObject.getJSONArray("actions");
                            if (actionArray != null && !actionArray.isEmpty()) {
                                //遍历集合中每条详细信息，封装成单独的对象
                                for (int i = 0; i < actionArray.size(); i++) {
                                    JSONObject actionObj = new JSONObject();
                                    //action字段
                                    actionObj.put("action", actionArray.getJSONObject(i));
                                    //common字段
                                    actionObj.put("common", commonObj);
                                    //page字段
                                    actionObj.put("page", pageObj);
                                    //由于action中已经有ts，不用再给ts字段了
                                    //写到动作日志的侧输出流中
                                    context.output(actionTag, actionObj.toJSONString());
                                }
                                //移除actions字段
                                jsonObject.remove("actions");
                            }

                            //~~~曝光日志~~~
                            //尝试获取曝光字段
                            JSONArray displayArray = jsonObject.getJSONArray("displays");
                            if (displayArray != null && !displayArray.isEmpty()) {
                                //遍历
                                for (int i = 0; i < displayArray.size(); i++) {
                                    JSONObject displayObj = new JSONObject();
                                    //display字段
                                    displayObj.put("display", displayArray.getJSONObject(i));
                                    //common字段
                                    displayObj.put("common", commonObj);
                                    //page字段
                                    displayObj.put("page", pageObj);
                                    //ts字段
                                    displayObj.put("ts", ts);
                                    //写出到侧输出流
                                    context.output(displayTag, displayObj.toJSONString());
                                }
                                //移除displays字段
                                jsonObject.remove("displays");
                            }
                            //该移除的字段都移除完了，直接写出主流即可
                            collector.collect(jsonObject.toJSONString());
                        }
                    }
                }
        );

        //使用标签获取5个侧输出流
        SideOutputDataStream<String> errDs = pageDs.getSideOutput(errTag);
        SideOutputDataStream<String> startDs = pageDs.getSideOutput(startTag);
        SideOutputDataStream<String> displayDs = pageDs.getSideOutput(displayTag);
        SideOutputDataStream<String> actionDs = pageDs.getSideOutput(actionTag);
        SideOutputDataStream<String> playDs = pageDs.getSideOutput(playTag);

        //测试
//        errDs.print("errDs");
//        startDs.print("startDs");
//        displayDs.print("displayDs");
//        actionDs.print("actionDs");
//        pageDs.print("pageDs");
//        playDs.print("pageDs");

        //将6个分流写入kafka对应主题
        pageDs.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        errDs.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        displayDs.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        actionDs.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        startDs.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        playDs.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PLAY));
    }
}
